package com.catmiao.spark.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: Spark01_Req1_HotCategoryTop10Analysis
 * @projectName spark_study
 * @description: 存在大量的shuffle操作，聚合算子，spark会提供优化，缓存
 * @author ChengMiao
 * @date 2024/3/11 23:14
 */
object Spark01_Req1_HotCategoryTop10Analysis_02 {

  def main(args: Array[String]): Unit = {


    // TODO 1. top10 热门品类
    val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")

    val sc = new SparkContext(conf)

    /**
     * 1. 读取原始的日志数据
     * 2. 统计品类的点击数量： (品类id,点击数量)
     * 3. 统计品类的下单数量： (品类id,下单数量)
     * 4. 统计品类的支付数量： (品类id,支付数量)
     *
     * 5. 排序品类，获取前10名
     *    - 点击数排序、下单数排序、支付数排序
     *    - 通过元组排序实现 (品类id,(点击数，下单数，支付数))
     *      6. 打印数据
     */

    // 1. 读取原始的日志数据
    val rdd = sc.textFile("datas/req_record/user_visit_action.txt")

    // 转换数据结构
    val flatRdd: RDD[(String, (Int, Int, Int))] = rdd.flatMap(
      item => {
        val datas = item.split("_")

        // 点击
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
        }

        // 下单
        else if (datas(8) != "null") {
          val cids = datas(8).split(",")
          cids.map(id => {
            (id, (0, 1, 0))
          })
        }

        // 支付
        else if (datas(10) != "null") {
          val cids = datas(10).split(",")
          cids.map(id => {
            (id, (0, 0, 1))
          })
        }

        else {
          Nil
        }

      }
    )

    val result: RDD[(String, (Int, Int, Int))] = flatRdd.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )

    // 排序
    val tuples: Array[(String, (Int, Int, Int))] = result.sortBy(_._2, false).take(10)

    tuples.foreach(println)
    sc.stop()
  }

}
